package cn.itcast.streaming.function.window;

import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.bean.TripModel;
import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.LinkedList;

//驾驶行程业务逻辑开发，低速，中速，高速分析车主驾驶行为
public class TripDriveWindowFunction extends RichWindowFunction<ItcastDataObj, TripModel,String, TimeWindow> {

    /**
     * 自定义窗口操作
     * @param s            分组字段
     * @param timeWindow   窗口操作
     * @param iterable     某辆车所以驾驶行程记录
     * @param collector    收集器
     * @throws Exception
     */
    @Override //
    public void apply(String s, TimeWindow timeWindow, Iterable<ItcastDataObj> iterable, Collector<TripModel> collector) throws Exception {
        //1、获取到某个车辆驾驶行程中所以的驾驶行为数据并进行转换
        LinkedList<ItcastDataObj> objList = Lists.newLinkedList(iterable);
        //2、对转换后的数据按照时间排序
        objList.sort(((o1, o2) -> {
            if(o1.getTerminalTimeStamp()>o2.getTerminalTimeStamp()){
                return 1;
            }else if(o1.getTerminalTimeStamp()<o2.getTerminalTimeStamp()){
                return -1;
            }else {
                return 0;
            }
        }));

        TripModel tripModel = getTripModel(objList);
        collector.collect(tripModel);//返回分析结果
    }

    private TripModel getTripModel(LinkedList<ItcastDataObj> itcastDataObjArrays){
        TripModel tripModel = new TripModel();
        //3、获取第一条数据
        ItcastDataObj firstObj = itcastDataObjArrays.getFirst();
        tripModel.setVin(firstObj.getVin());//vin都一样
        tripModel.setTripStartTime(firstObj.getTerminalTime());//行程开始时间
        tripModel.setStart_BMS_SOC(firstObj.getSoc());//开始剩余电量百分比
        tripModel.setStart_longitude(firstObj.getLng());//开始经度
        tripModel.setStart_latitude(firstObj.getLat()); //开始维度
        tripModel.setStart_mileage(firstObj.getTotalOdometer());//行程开始里程数
        //4、获取最后一条数据
        ItcastDataObj lastObj = itcastDataObjArrays.getLast();
        tripModel.setTripEndTime(lastObj.getTerminalTime());//行程结束时间
        tripModel.setEnd_BMS_SOC(lastObj.getSoc());//结束剩余电量百分比
        tripModel.setEnd_longitude(lastObj.getLng());//结束经度
        tripModel.setEnd_latitude(lastObj.getLat()); //结束维度
        tripModel.setEnd_mileage(lastObj.getTotalOdometer());//行程结束里程数

        //行程里程消耗
        tripModel.setMileage(lastObj.getTotalOdometer()-firstObj.getTotalOdometer());
        //行程时间消耗
        tripModel.setTime_comsuption(Math.abs(lastObj.getTerminalTimeStamp()-firstObj.getTerminalTimeStamp())/1000/60D);
        //上次报文soc ：下次行程开始的剩余电量和上次行程结束的剩余电量相等
        tripModel.setLastSoc(firstObj.getSoc()+0D);
        //上次报文的里程数
        tripModel.setLastMileage(firstObj.getTotalOdometer());
        //本次消耗的soc
        tripModel.setSoc_comsuption(Math.abs(lastObj.getSoc()-tripModel.getLastSoc()));
        //5、循环遍历集合中的所有数据（计算驾驶行为分析
        for (ItcastDataObj obj : itcastDataObjArrays) {
            //获取到每条数据的速度
            Double speed = obj.getSpeed();
            //获取上次报文的soc
            Double lastSoc = tripModel.getLastSoc();
            //如果当前车速小于150并且大于javabean保存的最大车速则修改javabean存储的最大车速
            if(tripModel.getMax_speed()<150 && tripModel.getMax_speed()<speed){
                //修改单词行程的最大车速，如果超过150则为异常数据
                tripModel.setMax_speed(speed);
            }
            //低速行驶
            if(speed>0 && speed <40){
                //低速行驶个数,,低速里程
                tripModel.setTotal_low_speed_nums(tripModel.getTotal_low_speed_nums()+1);
                //低速soc消耗(累计油耗+（最后一条数据的低速油耗-本次低速油耗））
                tripModel.setLow_BMS_SOC(tripModel.getLow_BMS_SOC()+Math.abs(tripModel.getLastSoc()-obj.getSoc()));
                //低速里程（累计低速里程+（本次低速里程-最后一条数据的低速里程））
                tripModel.setLow_BMS_Mileage(tripModel.getLow_BMS_Mileage()+Math.abs(obj.getTotalOdometer()-tripModel.getLastMileage()));
            }else if(speed>=40 && speed <80){
                //高速行驶个数,高速soc消耗,高速里程
                tripModel.setTotal_medium_speed_nums(tripModel.getTotal_medium_speed_nums()+1);
                //低速soc消耗(累计油耗+（最后一条数据的低速油耗-本次低速油耗））
                tripModel.setMedium_BMS_SOC(tripModel.getMedium_BMS_SOC()+Math.abs(tripModel.getLastSoc()-obj.getSoc()));
                //低速里程（累计低速里程+（本次低速里程-最后一条数据的低速里程））
                tripModel.setMedium_BMS_Mileage(tripModel.getMedium_BMS_Mileage()+Math.abs(obj.getTotalOdometer()-tripModel.getLastMileage()));
            }else if(speed>=80 && speed<150){
                //高速行驶个数,高速soc消耗,高速里程
                tripModel.setTotal_high_speed_nums(tripModel.getTotal_high_speed_nums()+1);
                //低速soc消耗(累计油耗+（最后一条数据的低速油耗-本次低速油耗））
                tripModel.setHigh_BMS_SOC(tripModel.getHigh_BMS_SOC()+Math.abs(tripModel.getLastSoc()-obj.getSoc()));
                //低速里程（累计低速里程+（本次低速里程-最后一条数据的低速里程））
                tripModel.setHigh_BMS_Mileage(tripModel.getHigh_BMS_Mileage()+Math.abs(obj.getTotalOdometer()-tripModel.getLastMileage()));
            }
            //修改上次的Soc
            tripModel.setLastSoc(obj.getSoc()+0D);
            //修改上次里程
            tripModel.setLastMileage(obj.getTotalOdometer()+0D);
        }
        //是否为异常数据
        if(itcastDataObjArrays.size()>1){
            tripModel.setTripStatus(0); //正常行程
        }else {
            tripModel.setTripStatus(1); //异常行程
        }
        //6、将分析结果返回
        return tripModel;
    }
}
